package com.joker.demo.admin.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.joker.demo.admin.domain.po.ExecutorStatusPO;
import com.joker.demo.admin.domain.po.JobGroupPO;
import com.joker.demo.admin.domain.po.JobInfoPO;
import com.joker.demo.admin.domain.po.JobRegistryPO;
import com.joker.demo.admin.mapper.ExecutorStatusMapper;
import com.joker.demo.admin.mapper.JobGroupMapper;
import com.joker.demo.admin.mapper.JobInfoMapper;
import com.joker.demo.admin.mapper.JobRegistryMapper;
import com.joker.demo.admin.service.ExecutorCleanupService;
import com.joker.demo.core.domain.enums.JobStatusEnum;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * 执行器清理服务实现类
 *
 * @author : feixiang.li
 * @since : 2025-01-27 10:00
 */
@Service
@Slf4j
public class ExecutorCleanupServiceImpl implements ExecutorCleanupService {

    @Resource
    private ExecutorStatusMapper executorStatusMapper;

    @Resource
    private JobRegistryMapper jobRegistryMapper;

    @Resource
    private JobGroupMapper jobGroupMapper;

    @Resource
    private JobInfoMapper jobInfoMapper;

    /**
     * 定时清理离线执行器任务
     * 每5分钟执行一次
     */
    @Scheduled(fixedRate = 300000) // 5分钟 = 300000毫秒
    public void scheduledCleanOfflineExecutors() {
        log.info("开始执行定时清理离线执行器任务...");
        checkAndCleanOfflineExecutors();
        log.info("定时清理离线执行器任务执行完成");
    }

    /**
     * 应用启动时初始化清理任务
     */
    @PostConstruct
    public void initCleanupTask() {
        log.info("开始执行应用启动时的执行器清理任务...");
        checkAndCleanOfflineExecutors();
        log.info("应用启动时的执行器清理任务执行完成");
    }

    @Override
    public void checkAndCleanOfflineExecutors() {
        try {
            log.info("开始检查离线执行器状态...");
            
            // 1. 清理t_job_executor_status表中的离线执行器
            cleanOfflineExecutorsFromStatus();
            
            // 2. 清理t_job_registry表中的离线执行器
            cleanOfflineExecutorsFromRegistry();
            
            // 3. 更新JobInfo表中的triggerStatus
            updateJobInfoTriggerStatus();
            
            log.info("离线执行器清理完成");
            
        } catch (Exception e) {
            log.error("检查并清理离线执行器失败", e);
        }
    }

    /**
     * 清理t_job_executor_status表中的离线执行器
     */
    private void cleanOfflineExecutorsFromStatus() {
        try {
            log.info("开始清理t_job_executor_status表中的离线执行器...");
            
            // 计算1分钟前的时间
            LocalDateTime oneMinuteAgo = LocalDateTime.now().minusMinutes(1);
            
            // 查询超过1分钟未心跳的执行器
            LambdaQueryWrapper<ExecutorStatusPO> queryWrapper = new LambdaQueryWrapper<>();
            queryWrapper.lt(ExecutorStatusPO::getLastHeartbeatTime, oneMinuteAgo);
            queryWrapper.eq(ExecutorStatusPO::getStatus, "ONLINE");
            
            List<ExecutorStatusPO> offlineExecutors = executorStatusMapper.selectList(queryWrapper);
            
            if (offlineExecutors.isEmpty()) {
                log.info("t_job_executor_status表中没有发现离线执行器");
                return;
            }
            
            log.info("t_job_executor_status表中发现 {} 个离线执行器，开始清理...", offlineExecutors.size());
            
            for (ExecutorStatusPO executor : offlineExecutors) {
                String registryKey = executor.getRegisterKey();
                log.info("处理t_job_executor_status表中的离线执行器: {}", registryKey);
                
                // 更新执行器状态为离线
                executor.setStatus("OFFLINE");
                executorStatusMapper.updateById(executor);
                log.info("执行器 {} 状态已更新为离线", registryKey);
                
                // 从JobGroup表中删除地址列表
                removeFromJobGroup(registryKey);
            }
            
            log.info("t_job_executor_status表中的离线执行器清理完成");
            
        } catch (Exception e) {
            log.error("清理t_job_executor_status表中的离线执行器失败", e);
        }
    }

    @Override
    public void cleanOfflineExecutorsFromRegistry() {
        try {
            log.info("开始清理t_job_registry表中的离线执行器...");
            
            // 获取所有注册的执行器
            List<JobRegistryPO> allRegistries = jobRegistryMapper.selectList(null);
            
            if (allRegistries.isEmpty()) {
                log.info("t_job_registry表中没有注册的执行器");
                return;
            }
            
            // 获取所有注册Key
            Set<String> registryKeys = allRegistries.stream()
                    .map(JobRegistryPO::getRegistryKey)
                    .collect(Collectors.toSet());
            
            log.info("t_job_registry表中发现 {} 个注册的执行器", registryKeys.size());
            
            // 检查每个注册Key是否在t_job_executor_status表中在线
            for (String registryKey : registryKeys) {
                LambdaQueryWrapper<ExecutorStatusPO> statusWrapper = new LambdaQueryWrapper<>();
                statusWrapper.eq(ExecutorStatusPO::getRegisterKey, registryKey);
                statusWrapper.eq(ExecutorStatusPO::getStatus, "ONLINE");
                
                ExecutorStatusPO executorStatus = executorStatusMapper.selectOne(statusWrapper);
                
                if (executorStatus == null) {
                    log.info("执行器 {} 在t_job_executor_status表中不在线，开始清理...", registryKey);
                    
                    // 删除注册信息
                    LambdaQueryWrapper<JobRegistryPO> registryWrapper = new LambdaQueryWrapper<>();
                    registryWrapper.eq(JobRegistryPO::getRegistryKey, registryKey);
                    int deletedRegistries = jobRegistryMapper.delete(registryWrapper);
                    log.info("执行器 {} 注册信息已删除，删除数量: {}", registryKey, deletedRegistries);
                    
                    // 从JobGroup表中删除地址列表
                    removeFromJobGroup(registryKey);
                } else {
                    log.debug("执行器 {} 在t_job_executor_status表中在线，无需清理", registryKey);
                }
            }
            
            log.info("t_job_registry表中的离线执行器清理完成");
            
        } catch (Exception e) {
            log.error("清理t_job_registry表中的离线执行器失败", e);
        }
    }
    
    /**
     * 从JobGroup表中删除地址列表
     */
    private void removeFromJobGroup(String registryKey) {
        try {
            // 查询所有JobGroup
            List<JobGroupPO> jobGroups = jobGroupMapper.selectList(null);
            
            for (JobGroupPO jobGroup : jobGroups) {
                if (StringUtils.isNotBlank(jobGroup.getAddressList()) && StringUtils.contains(jobGroup.getAddressList(), registryKey)) {
                    // 移除该地址
                    jobGroup.removeAddressList(registryKey);
                    jobGroupMapper.updateById(jobGroup);
                    log.info("从JobGroup {} 中移除执行器地址: {}", jobGroup.getGroupJobName(), registryKey);
                }
            }
        } catch (Exception e) {
            log.error("从JobGroup中移除执行器地址失败: {}", registryKey, e);
        }
    }
    
    @Override
    public void updateJobInfoTriggerStatus() {
        try {
            log.info("开始更新JobInfo表中的triggerStatus...");
            
            // 查询所有JobInfo
            List<JobInfoPO> jobInfoList = jobInfoMapper.selectList(null);
            
            for (JobInfoPO jobInfo : jobInfoList) {
                boolean hasValidExecutor = false;
                
                // 检查JobGroup中是否有有效的执行器
                if (jobInfo.getGroupName() != null) {
                    LambdaQueryWrapper<JobGroupPO> groupWrapper = new LambdaQueryWrapper<>();
                    groupWrapper.eq(JobGroupPO::getGroupJobName, jobInfo.getGroupJobName());
                    JobGroupPO jobGroup = jobGroupMapper.selectOne(groupWrapper);
                    
                    if (jobGroup != null && jobGroup.getAddressList() != null && !jobGroup.getAddressList().trim().isEmpty()) {
                        // 检查地址列表中的执行器是否在线
                        String[] addresses = jobGroup.getAddressList().split(",");
                        for (String address : addresses) {
                            String trimmedAddress = address.trim();
                            if (!trimmedAddress.isEmpty()) {
                                LambdaQueryWrapper<ExecutorStatusPO> executorWrapper = new LambdaQueryWrapper<>();
                                executorWrapper.eq(ExecutorStatusPO::getRegisterKey, trimmedAddress);
                                executorWrapper.eq(ExecutorStatusPO::getStatus, "ONLINE");
                                
                                ExecutorStatusPO executorStatus = executorStatusMapper.selectOne(executorWrapper);
                                if (executorStatus != null) {
                                    hasValidExecutor = true;
                                    break;
                                }
                            }
                        }
                    }
                }
                
                // 更新triggerStatus
                int newTriggerStatus = hasValidExecutor ? JobStatusEnum.READY.getCode() : JobStatusEnum.EXCEPTION.getCode();
                if (jobInfo.getTriggerStatus() == null || !jobInfo.getTriggerStatus().equals(newTriggerStatus)) {
                    jobInfo.setTriggerStatus(newTriggerStatus);
                    jobInfoMapper.updateById(jobInfo);
                    
                    JobStatusEnum statusEnum = JobStatusEnum.getByCode(newTriggerStatus);
                    String statusDesc = statusEnum != null ? statusEnum.getDesc() : "未知状态";
                    log.info("JobInfo {} 的triggerStatus已更新为: {} ({})", 
                            jobInfo.getJobName(), 
                            newTriggerStatus, 
                            statusDesc);
                }
            }
            
            log.info("JobInfo表中的triggerStatus更新完成");
            
        } catch (Exception e) {
            log.error("更新JobInfo triggerStatus失败", e);
        }
    }
}
